package producer

import (
	"git.edaijia-inc.cn/edaijia/ego/elogger"
	eloggerkafak "git.edaijia-inc.cn/edaijia/ego/elogger/kafka"

	"github.com/Shopify/sarama"
)

type EKafkaProducerConfig struct {
	Brokers []string `ini:"brokers"` //kafka broker地址列表
	Version string   `ini:"version"`
	Verbose bool     `ini:"verbose"`
}

type EKafkaProducer struct {
	Producer sarama.SyncProducer
	Conf     *EKafkaProducerConfig
}

func NewEKafkaProducer(conf *EKafkaProducerConfig) (*EKafkaProducer, error) {
	var err error

	//默认0.8.2.1
	if conf.Version == "" {
		conf.Version = "0.8.2.1"
	}

	eloggerkafak.InitELoggerKafka(conf.Verbose)

	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Retry.Max = 3
	config.Producer.Return.Successes = true
	config.Version, err = sarama.ParseKafkaVersion(conf.Version)
	if err != nil {
		elogger.Error("Kafka version error:", err)
		return nil, err
	}

	producer, err := sarama.NewSyncProducer(conf.Brokers, config)
	if err != nil {
		elogger.Error("Failed to start Kafka producer:", err)
		return nil, err
	}

	return &EKafkaProducer{producer, conf}, nil
}

func (p *EKafkaProducer) SendMessage(topic string, payload string) (int32, int64, error) {
	if p.Producer == nil {
		elogger.Error("Kafka Producer is nil, call NewEKafkaProducer to init firstly")
	}

	msg := &sarama.ProducerMessage{Topic: topic, Value: sarama.StringEncoder(payload)}
	return p.Producer.SendMessage(msg)
}
